package com.klun.hdfs.hd.recordwriter;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class FilterRecordWriter extends RecordWriter<Text, NullWritable> {

    FSDataOutputStream atguiguOut = null;
    FSDataOutputStream otherOut = null;

    public FilterRecordWriter() {
    }
    public FilterRecordWriter(TaskAttemptContext job) {
        // 获取文件系统
        FileSystem fs;
        try{
            fs = FileSystem.get(job.getConfiguration());

            Path guiguPath = new Path("e:/atguigu.log");
            Path otherPath = new Path("e:/other.log");

            atguiguOut = fs.create(guiguPath);
            otherOut = fs.create(otherPath);
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        if(key.toString().contains("atguigu")){
            atguiguOut.write(key.toString().getBytes());
        }else {
            otherOut.write(key.toString().getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {

        if(atguiguOut!=null){
            atguiguOut.close();
        }
        if(otherOut != null){
            otherOut.close();
        }
    }
}
